package com.zk.storm;
//用于提取关键数据,同时转化时间格式
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

public class DataParseBolt extends BaseRichBolt {
    private OutputCollector collector;
    private ObjectMapper jsonMapper;
    private SimpleDateFormat dateFormat;

    @Override
    public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        this.jsonMapper = new ObjectMapper();
        this.dateFormat = new SimpleDateFormat("yyyyMMdd");
    }

    @Override
    public void execute(Tuple input) {
        try {
            String jsonStr = input.getValueByField("original_message").toString();
            String country = input.getStringByField("country");
            JsonNode json = jsonMapper.readTree(jsonStr);

            // 解析原始字段
            String rawDate = json.get("lastCounter").asText();
            int hits = json.get("hits").asInt();
            String agent = json.get("agent").asText();
            String platform = json.get("platform").asText();
            String device = json.get("device").asText();

            // 转换日期格式
            Date date = new SimpleDateFormat("yyyy/MM/dd").parse(rawDate);
            String dateKey = dateFormat.format(date);

            // 输出需要的字段
            collector.emit(new Values(
                    dateKey,
                    hits,
                    agent,
                    platform,
                    device,
                    country
            ));
            collector.ack(input);
        } catch (Exception e) {
            collector.fail(input);
            throw new RuntimeException("数据解析失败: " + e.getMessage());
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 声明扩展后的输出字段
        declarer.declare(new Fields(
                "date_key",
                "hits",
                "agent",
                "platform",
                "device",
                "country"
        ));
    }
}